package com.rrd.hive.udf;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.LinkedHashMap;
import java.util.List;

import com.alibaba.fastjson.JSONException;
import com.alibaba.fastjson.JSONObject;
import org.apache.hadoop.hive.ql.exec.Description;
import org.apache.hadoop.hive.ql.exec.UDFArgumentException;
import org.apache.hadoop.hive.ql.metadata.HiveException;
import org.apache.hadoop.hive.ql.udf.generic.GenericUDF;
import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector;
import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspectorConverters;
import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspectorFactory;
import org.apache.hadoop.hive.serde2.objectinspector.PrimitiveObjectInspector;
import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector.Category;
import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspectorConverters.Converter;
import org.apache.hadoop.hive.serde2.objectinspector.primitive.PrimitiveObjectInspectorFactory;
import org.apache.hadoop.hive.serde2.objectinspector.primitive.PrimitiveObjectInspectorUtils;
import org.apache.hadoop.hive.serde2.objectinspector.primitive.PrimitiveObjectInspectorUtils.PrimitiveGrouping;

@Description(
        name = "json_to_map",
        value = "_FUNC_(text) - Creates a map by json string ",
        extended = "parse json string to map"
)
@Deprecated
public class GenericUDFJsonToMap extends GenericUDF {
    private final LinkedHashMap<Object, Object> ret = new LinkedHashMap();
    private transient Converter soi_text;
    private transient Converter soi_text1;

    public GenericUDFJsonToMap() {
    }

    public ObjectInspector initialize(ObjectInspector[] arguments) throws UDFArgumentException {
        for(int idx = 0; idx < Math.min(arguments.length, 2); ++idx) {
            if (arguments[idx].getCategory() != Category.PRIMITIVE || PrimitiveObjectInspectorUtils.getPrimitiveGrouping(((PrimitiveObjectInspector)arguments[idx]).getPrimitiveCategory()) != PrimitiveGrouping.STRING_GROUP) {
                throw new UDFArgumentException("All argument should be string/character type");
            }
        }

        this.soi_text = ObjectInspectorConverters.getConverter(arguments[0], PrimitiveObjectInspectorFactory.javaStringObjectInspector);
        if (arguments.length >= 2) {
            this.soi_text1 = ObjectInspectorConverters.getConverter(arguments[1], PrimitiveObjectInspectorFactory.javaStringObjectInspector);
        }
        return ObjectInspectorFactory.getStandardMapObjectInspector(PrimitiveObjectInspectorFactory.javaStringObjectInspector, PrimitiveObjectInspectorFactory.javaStringObjectInspector);
    }

    public Object evaluate(DeferredObject[] arguments) throws HiveException {
        this.ret.clear();
        String jsonStr = (String)this.soi_text.convert(arguments[0].get());
        String exclude = this.soi_text1 == null ? ":" : (String)this.soi_text1.convert(arguments[1].get());
        List<String> excludes = new ArrayList<>();
        if (exclude != null){
            excludes = Arrays.asList(exclude.split(","));
        }

        try{
            JSONObject jsonObj = JSONObject.parseObject(jsonStr);
            for(String key: jsonObj.keySet()){
                if (excludes.size() == 0 || !excludes.contains(key)){
                    String val = jsonObj.getString(key);
                    this.ret.put(key, val);
                }
            }
        }catch (JSONException ignored) {
        }
        return this.ret;
    }

    public String getDisplayString(String[] children) {
        assert children.length <= 2;

        return this.getStandardDisplayString("json_to_map", children, ",");
    }
}
